package quickstart;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;

import java.util.HashMap;
import java.util.Map;

/**
 * @Author:RenPu
 * @Date: 2020/4/20 14:09
 * @Version: 1.0
 * @description:
 */
public class WordCountBolt extends BaseRichBolt {


    //
    private HashMap<String,Long> wordCount=null;

    private OutputCollector collector;


    /**
     * 准备方法
     * @param map
     * @param topologyContext
     * @param outputCollector
     */
    @Override
    public void prepare(Map<String, Object> map, TopologyContext topologyContext, OutputCollector outputCollector) {

        //初始化Map集合，用来统计单词出现的次数以及封装核算后的数据
        this.wordCount=new HashMap<String, Long>(20);

    }

    /**
     * 执行方法
     * @param tuple
     */
    @Override
    public void execute(Tuple tuple) {

        //通过元祖信息的唯一标识，获取对应组件的数据
        String word = tuple.getStringByField("word");

        //类记单词出现的次数
        Long num = wordCount.getOrDefault(word, 0L);

        //计数累加
        num++;

        //将数据封装到map里
        wordCount.put(word,num);

        System.out.println(word+"\t"+num);


        /**
         * 无须锚定因为没有产生新的bolt
         */
        collector.ack(tuple);


    }


    /**
     * 声明元祖对象信息的方法
     * @param outputFieldsDeclarer
     */
    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

    }
}
